package com.zeus.base.smscenter.schedule.task;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.openservices.ons.api.*;
import com.google.common.base.Strings;
import com.smscenter.server.core.IExecutor;
import com.smscenter.server.core.ems.EmsExecutor;
import com.smscenter.server.core.response.EmsReponse;
import com.smscenter.server.core.response.VComcnReponse;
import com.smscenter.server.core.response.WXReponse;
import com.smscenter.server.core.response.WelinkReponse;
import com.smscenter.server.core.tradingcenter.TradingConsumExecutor;
import com.smscenter.server.core.vcomcn.VComSmsExecutor;
import com.smscenter.server.core.vcomcn.VComVoiceExecutor;
import com.smscenter.server.core.wechat.WXExecutor;
import com.smscenter.server.core.welink.WelinkExecutor;
import com.smscenter.server.core.ztsms.ZtSmsError;
import com.smscenter.server.core.ztsms.ZtSmsExecutor;
import com.zeus.base.privilege.model.TbUser;
import com.zeus.base.privilege.service.IUserService;
import com.zeus.base.privilege.utils.*;
import com.zeus.base.smscenter.common.constant.SmsConstants;
import com.zeus.base.smscenter.common.security.MD5;
import com.zeus.base.smscenter.common.util.StringUtils;
import com.zeus.base.smscenter.model.SmsLogInfo;
import com.zeus.base.smscenter.model.SmsSupplierProperty;
import com.zeus.base.smscenter.model.Tenant;
import com.zeus.base.smscenter.service.ISmsLogInfoService;
import com.zeus.base.smscenter.service.ISmsSettingService;
import com.zeus.base.smscenter.service.ISupplierPropertyService;
import com.zeus.base.tradingcenter.model.ConsumeType;
import com.zeus.base.tradingcenter.service.IConsumeDetailService;
import com.zeus.base.tradingcenter.service.IConsumeTypeService;
import org.apache.log4j.Logger;
import org.joda.time.DateTime;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;
import java.util.*;

/**
 * 短信消息消费者
 */
public class SmsOnsMessageListener {

    Logger logger = Logger.getLogger(this.getClass());


    @Autowired
    private IUserService iUserService;

    @Autowired
    private ISmsLogInfoService smsLogInfoService;

    @Autowired
    private ISmsSettingService tenantService;

    @Autowired
    private ISupplierPropertyService smsSupplierPropertyService;

    @Autowired
    private IConsumeTypeService iConsumeTypeService;
    @Autowired
    private IConsumeDetailService iConsumeDetailService;

    private String consumerId;

    private String accessKey;

    private String secretKey;

    private String topicName;

    private String tagName;

    private Consumer consumer;

    public void onStart() {
        Properties props = new Properties();
        props.put(PropertyKeyConst.ConsumerId, consumerId);
        props.put(PropertyKeyConst.AccessKey, accessKey);
        props.put(PropertyKeyConst.SecretKey, secretKey);
        consumer = ONSFactory.createConsumer(props);
        consumer.subscribe(topicName, tagName, new MyMessageListener());
        consumer.start();
        System.out.println("启动ONS消费者");
    }

    public ISmsLogInfoService getSmsLogInfoService() {
        return smsLogInfoService;
    }

    public void setSmsLogInfoService(ISmsLogInfoService smsLogInfoService) {
        this.smsLogInfoService = smsLogInfoService;
    }

    public ISupplierPropertyService getSmsSupplierPropertyService() {
        return smsSupplierPropertyService;
    }

    public void setSmsSupplierPropertyService(ISupplierPropertyService smsSupplierPropertyService) {
        this.smsSupplierPropertyService = smsSupplierPropertyService;
    }

    public ISmsSettingService getTenantService() {
		return tenantService;
	}

	public void setTenantService(ISmsSettingService tenantService) {
		this.tenantService = tenantService;
	}

	public String getConsumerId() {
        return consumerId;
    }

    public void setConsumerId(String consumerId) {
        this.consumerId = consumerId;
    }

    public String getAccessKey() {
        return accessKey;
    }

    public void setAccessKey(String accessKey) {
        this.accessKey = accessKey;
    }

    public String getSecretKey() {
        return secretKey;
    }

    public void setSecretKey(String secretKey) {
        this.secretKey = secretKey;
    }

    public String getTopicName() {
        return topicName;
    }

    public void setTopicName(String topicName) {
        this.topicName = topicName;
    }

    public String getTagName() {
        return tagName;
    }

    public void setTagName(String tagName) {
        this.tagName = tagName;
    }

    public void onStop() {
        consumer.shutdown();
        System.out.println("关闭ONS消费者");
    }

    private class MyMessageListener implements MessageListener {
        @Override
        public Action consume(Message message, ConsumeContext consumeContext) {
        	JSONObject obj = JSON.parseObject(new String(message.getBody()));
        	obj.put("messageId", message.getMsgID());
            boolean action = onMessage(obj.toJSONString());
            return action ? Action.CommitMessage : Action.ReconsumeLater;
        }

    }


    public synchronized boolean  onMessage(String message) {
        boolean action=true;
        Long start=System.currentTimeMillis();
        
        try {
            logger.info("smsmessage:"+message);
            SmsLogInfo smsLogInfo = JSON.parseObject(message, SmsLogInfo.class);//校验数据格式
            smsLogInfo.setAuthCenterUserName(smsLogInfo.getTenant());
            if (smsLogInfo.getSupplierChannel() != null && !"4".equals(smsLogInfo.getSupplierChannel().toString())) {
                smsLogInfo.setSupplierChannel(null);
            }
            Map<String, Object> params = new HashMap<String, Object>();
            params.put("code", smsLogInfo.getTenant());
            params.put("supplierNo", smsLogInfo.getSupplierChannel());
            params.put("appPlatform", smsLogInfo.getAppPlatform());
          
            if (smsLogInfo.getSupplierChannel() == null || "".equals(smsLogInfo.getSupplierChannel())) {//如果没有传通道字段走用户默认通道
            	
            	Tenant qt = new Tenant();
            	qt.setCode(smsLogInfo.getTenant());
            	Tenant rst = tenantService.findByCode(qt);
            	if (rst != null) {
            		if ("SMS".equals(smsLogInfo.getSendType())) {
            			params.put("supplierNo", rst.getDefaultSmsSupplierNo());
            			smsLogInfo.setSupplierChannel(Integer.parseInt(rst.getDefaultSmsSupplierNo()));
                    } else {
                    	params.put("supplierNo", rst.getDefaultVoiceSupplierNo());
                    	smsLogInfo.setSupplierChannel(Integer.parseInt(rst.getDefaultVoiceSupplierNo()));
                    }
				}
			}
            logger.info(smsLogInfo.getAppPlatform()+"平台,"+smsLogInfo.getSupplierChannel());
            List<SmsSupplierProperty> smsSupplierPropertyList = smsSupplierPropertyService.find(params);
            
            Long end=System.currentTimeMillis();
            
            logger.info("查询数据库时间cost:"+(start-end)/1000L);
            params.put("content", smsLogInfo.getContent());
            params.put("phone", smsLogInfo.getPhone());
            params.put("bizNo", smsLogInfo.getBizNo());

            boolean ifSendMessage = true;
            boolean ifUpdate = true;

            for (SmsSupplierProperty smsSupplierProperty : smsSupplierPropertyList) {
                params.put(smsSupplierProperty.getPropertyName(), smsSupplierProperty.getPropertyValue());
            }
            String timeExpireParam = (String) params.get("timeExpire");//失效时间
            int timeExpire = 3600 * 24;//默认失效时间为24小时
            if (!Strings.isNullOrEmpty(timeExpireParam)) {
                timeExpire = Integer.parseInt(params.get("timeExpire").toString());
            }
            smsLogInfo.setBeginDate(new DateTime().minusSeconds(timeExpire).toDate());
            smsLogInfo.setEndDate(new DateTime().plusDays(1).toDate());//默认增加一天的时间差
            smsLogInfo.setHashContent(getHashContent(smsLogInfo));
            SmsLogInfo temp = null;
            if (smsSupplierPropertyList == null || smsSupplierPropertyList.size() == 0) {
                ifSendMessage = false;
                ifUpdate = false;
                smsLogInfo.setOutResponseMessage("客户配置信息有误，查不到相关信息");
                smsLogInfo.setStatus(SmsConstants.FAIL);
                smsLogInfo.setOutResponseCode(SmsConstants.CONFIGURATION_ERROR);
            } else if ("SMS".equals(smsLogInfo.getSendType()) && smsLogInfo.getPhone().length() != 11) {
                ifSendMessage = false;
                ifUpdate = false;
                smsLogInfo.setOutResponseMessage("接收手机号码格式不正确");
                if (smsLogInfo.getPhone().length() > 20) {
                    smsLogInfo.setPhone(smsLogInfo.getPhone().substring(0,20));
                }
                smsLogInfo.setStatus(SmsConstants.FAIL);
                smsLogInfo.setOutResponseCode(SmsConstants.CONFIGURATION_ERROR);
            } else if ("N".equals(smsLogInfo.getIsValidate())) {
            	ifSendMessage = true;
                ifUpdate = false;
            } else{
            	temp = smsLogInfoService.findByTenantAndHashContentAndCreateDate(smsLogInfo);
            	logger.info("查询数据库时间去重cost:"+(start-System.currentTimeMillis())/1000L);
                if (temp == null) {  //发送短信，插入新短信内容
                    ifSendMessage = true;
                    ifUpdate = false;
                } else {
                    if (temp.getStatus().equals(SmsConstants.FAIL)) {  //更新状态 发送短信
                        ifUpdate = true;
                        ifSendMessage = true;
                    } else {  //不发短信 删除队列内容
                        ifSendMessage = false;
                    }
                }
            }
            if (smsLogInfo.getSupplierChannel().equals(4)) {
                excuteMessageWX(ifSendMessage, ifUpdate, smsLogInfo, temp, params);
            } else {
                excuteMessage(ifSendMessage, ifUpdate, smsLogInfo, temp, params);
            }
        } catch (Exception e) {
            action = false;
            logger.error("处理失败:" + e + "失败的消息内容:"+message);
        }finally {
            return action;
        }
    }

    private void excuteMessage(boolean ifSendMessage, boolean ifUpdate, SmsLogInfo newSmsLogInfo, SmsLogInfo oldSmsLogInfo, Map<String, Object> params) throws IOException {
    	Long start=System.currentTimeMillis();
    	if(StringUtil.isEmpty(newSmsLogInfo.getAuthCenterUserName()) || StringUtil.isEmpty(newSmsLogInfo.getAuthCenterPassword())){
            ifSendMessage = false;
            newSmsLogInfo.setStatus(SmsConstants.FAIL);
            newSmsLogInfo.setTradingRemark("用户名或密码为空!");
        }
        TbUser user = iUserService.getUserByLoginName(newSmsLogInfo.getAuthCenterUserName());
        if (user == null || !newSmsLogInfo.getAuthCenterPassword().equals(user.getLoginPwd())) {
            ifSendMessage = false;
            newSmsLogInfo.setStatus(SmsConstants.FAIL);
            newSmsLogInfo.setTradingRemark("用户名密码不对!");
        }
        if (ifSendMessage) {//发送短信
            //发送短信前,先到交易中心进行扣费,如果扣费成功,则继续发送短信,否则不发送
            Double amount;
            if("VOICE".equals(newSmsLogInfo.getSendType())){
                amount = Double.parseDouble((String)params.get("voiceUnitPrice"));
            }else{
                amount = Double.parseDouble((String)params.get("unitPrice"));
            }
            //计算短信条数
            setContentLength(newSmsLogInfo, params.get("content")+""+(params.get("autograph") != null?params.get("autograph"):""));
            newSmsLogInfo.setAmount(new Double(ArithUtil.mul(ArithUtil.mul(amount , newSmsLogInfo.getUseSmsCount()),100)).intValue());
            //交易中心短信扣费接口
            TradingConsumExecutor tradingExecutor = new TradingConsumExecutor();
            ResultModel resultModelConsum = tradingExecutor.excuteConsume(getConsumeType("短信", iConsumeTypeService), ArithUtil.mul(amount , newSmsLogInfo.getUseSmsCount()), user, iConsumeDetailService);
            //扣费完成
            if(resultModelConsum.isSuccess()){
                try {
                    //如果扣费成功,则设置交易单号
                    newSmsLogInfo.setLoadStatus("2");//表示交易中心扣费完成
                    newSmsLogInfo.setTradingNo(resultModelConsum.getObject().toString());
                    if(newSmsLogInfo.getTradingNo()!=null && !"".equals(newSmsLogInfo.getTradingNo())){
                        params.put("bizNo", newSmsLogInfo.getTradingNo());
                    }else{
                        logger.error("当前短信失败,交易中心扣费成功后未返回交易单号!"+newSmsLogInfo.getBizNo());
                    }
                    IExecutor executor = null;
                    if ("SMS".equals(newSmsLogInfo.getSendType())) {
                        if(!"4".equals(newSmsLogInfo.getSupplierChannel()+"") && params.get("autograph") != null){
                            if("1".equals(newSmsLogInfo.getSupplierChannel()+"")){
                                if ("fytyjfzz".equals(params.get("loginName")+"")) {
                                    params.put("content", params.get("autograph")+(String)params.get("content"));
                                }
                            }else{
                                params.put("content", (String)params.get("content") + params.get("autograph"));
                            }
                        }
                        newSmsLogInfo.setContent((String)params.get("content"));
                        if("1".equals(newSmsLogInfo.getSupplierChannel()+"")) {//1：及时通,2:EMS,3:微网通联,4:微信推送,5:上海助通
                            executor = new VComSmsExecutor();
                        }else if("2".equals(newSmsLogInfo.getSupplierChannel()+"")) {
                            executor = new EmsExecutor();
                        }else if("3".equals(newSmsLogInfo.getSupplierChannel()+"")) {
                            executor = new WelinkExecutor();
                        }else if("4".equals(newSmsLogInfo.getSupplierChannel()+"")){
                            executor = new WXExecutor();
                        }else if("5".equals(newSmsLogInfo.getSupplierChannel()+"")){
                            executor = new ZtSmsExecutor();
                        }
                    } else {
                        //newSmsLogInfo.setContentLength(0);
                        //newSmsLogInfo.setUseSmsCount(1);
                        executor = new VComVoiceExecutor();
                    }
                    String response = executor.excute(params);
                    logger.info("发送短信结束"+(start-System.currentTimeMillis())/1000L);
                    newSmsLogInfo.setOutResponseCode(response);

                    String result = "";
                    if("1".equals(newSmsLogInfo.getSupplierChannel()+"")) {//1：及时通,2:EMS,3:微网通联,4:微信,5:上海助通
                        result = VComcnReponse.getInstance().getResponseMessage(response);
                    }else if("2".equals(newSmsLogInfo.getSupplierChannel()+"")) {
                        result = EmsReponse.getInstance().getResponseMessage(response);
                    }else if("3".equals(newSmsLogInfo.getSupplierChannel()+"")) {
                        result = WelinkReponse.getInstance().getResponseMessage(response);
                    }else if("4".equals(newSmsLogInfo.getSupplierChannel()+"")) {
                        result = WXReponse.getInstance().getResponseMessage(response);
                    }else if("5".equals(newSmsLogInfo.getSupplierChannel()+"")){
                        String returnCode = "";
                        String returnSmsId = "";
                        if (response.indexOf(",")>0){
                            String[] returnArr = response.split(",");
                            returnCode = returnArr[0];
                            returnSmsId = returnArr[1];
                            newSmsLogInfo.setMessagePlatformId(returnSmsId);
                        }else {
                            returnCode = response;
                        }
                        result = ZtSmsError.getErrorMess(Integer.parseInt(returnCode));
                    }
                    newSmsLogInfo.setOutResponseMessage(result);

                    if (SmsConstants.SUCCESS.equals(result)) {
                        newSmsLogInfo.setStatus(SmsConstants.SUCCESS);
                        newSmsLogInfo.setTradingRemark(resultModelConsum.getMessage());
                    } else {
                        newSmsLogInfo.setStatus(SmsConstants.FAIL);
                        //发送失败,对已扣短信费用回冲
                        ResultModel resultConsumRevert = tradingExecutor.excuteRevertConsume(newSmsLogInfo.getTradingNo(), user, null, iConsumeDetailService);
                        newSmsLogInfo.setTradingRemark(resultConsumRevert.getMessage());
                        if(resultConsumRevert.isSuccess()){
                            newSmsLogInfo.setLoadStatus("1");
                        }
                    }
                } catch (Exception e) {
                    if (StringUtil.isNotEmpty(newSmsLogInfo.getTradingNo())) {
                        tradingExecutor.excuteRevertConsume(newSmsLogInfo.getTradingNo(), user, null, iConsumeDetailService);
                    }
                    throw e;
                }
            }else{
                //扣费失败
                newSmsLogInfo.setStatus(SmsConstants.FAIL);
                newSmsLogInfo.setTradingRemark(resultModelConsum.getMessage());
            }
        }
        if (ifUpdate) {
            if (oldSmsLogInfo.getStatus().equals(SmsConstants.FAIL)) {
                newSmsLogInfo.setId(oldSmsLogInfo.getId());
                newSmsLogInfo.setLastUpdateTime(new Date());
                smsLogInfoService.update(newSmsLogInfo);
            }
        } else {
            if (newSmsLogInfo.getPhone().length() > 20) {
                newSmsLogInfo.setPhone(newSmsLogInfo.getPhone().substring(0,20));
            }
            newSmsLogInfo.setId(IdWorkerTools.getInstance().getId()+"");
            newSmsLogInfo.setLastUpdateTime(new Date());
            smsLogInfoService.save(newSmsLogInfo);
        }
        logger.info("发送短信全结束"+(start-System.currentTimeMillis())/1000L);
    }

    /**
     * 微信发送方法
     * @param ifSendMessage
     * @param ifUpdate
     * @param newSmsLogInfo
     * @param oldSmsLogInfo
     * @param params
     * @throws IOException
     */
    private void excuteMessageWX(boolean ifSendMessage, boolean ifUpdate, SmsLogInfo newSmsLogInfo, SmsLogInfo oldSmsLogInfo, Map<String, Object> params) throws IOException {
        if (ifSendMessage) {//发送短信
            IExecutor executor = new WXExecutor();
            String response = executor.excute(params);
            newSmsLogInfo.setOutResponseCode(response);
            String result = WXReponse.getInstance().getResponseMessage(response);
            newSmsLogInfo.setOutResponseMessage(result);
            if (SmsConstants.SUCCESS.equals(result)) {
                newSmsLogInfo.setStatus(SmsConstants.SUCCESS);
            } else {
                newSmsLogInfo.setStatus(SmsConstants.FAIL);
            }
        }
        if (ifUpdate) {
            if (oldSmsLogInfo.getStatus().equals(SmsConstants.FAIL)) {
                newSmsLogInfo.setId(oldSmsLogInfo.getId());
                newSmsLogInfo.setLastUpdateTime(new Date());
                smsLogInfoService.update(newSmsLogInfo);
            }
        } else {
            if (StringUtil.isNotEmpty(newSmsLogInfo.getPhone()) && newSmsLogInfo.getPhone().length() > 20) {
                newSmsLogInfo.setPhone(newSmsLogInfo.getPhone().substring(0,20));
            }
            newSmsLogInfo.setId(IdWorkerTools.getInstance().getId()+"");
            newSmsLogInfo.setLastUpdateTime(new Date());
            newSmsLogInfo.setAmount(1);
            smsLogInfoService.save(newSmsLogInfo);
        }
    }

    private String getHashContent(SmsLogInfo smsLogInfo) {
        StringBuffer sb = new StringBuffer();
        sb.append(smsLogInfo.getTenant()).//租户
                append(smsLogInfo.getAppPlatform()).//平台
                append(smsLogInfo.getBizNo()).append(smsLogInfo.getBizType()).
                append(smsLogInfo.getPhone()).//电话
                append(smsLogInfo.getContent());//内容
        return MD5.generate(sb.toString(), false);
    }
    
    /**
     * 计算短信字数与所占条数
     * @param smsLogInfo
     * @param content
     * 70个字符为一条计费，最多支持500个字符，实际支持的最大字符根据通道方限制为准(一般是300—500个字符不等)。超过70个字按67个字符一条计费，
     * 例如给一个号码提交了一条137个字符的短信137=67*2+3超过了2条计费一个号码就会按3条计费。【70=1条；71=2条；134=2条；135=3条】
     * 
     * 国内所有短信使用均以70个字符一条计算
     * 长短信计算为：134=67+67，为两条计费。多条计费统一为：67+67+67+……+67。
     */
	private void setContentLength(SmsLogInfo smsLogInfo, String content) {
		if (!StringUtils.isEmpty(content)) {
			int l = content.length();
			smsLogInfo.setContentLength(l);
			if (l <= 70) {
				smsLogInfo.setUseSmsCount(1);
			}else{
				smsLogInfo.setUseSmsCount((int) Math.ceil(l/67.0));
			}
		}
	}


    private static HashMap<String, HashMap<String, Object>> cacheTypeMap = new HashMap<String, HashMap<String, Object>>();
    public static ConsumeType getConsumeType(String type, IConsumeTypeService iConsumeTypeService) {
        HashMap<String, Object> cacheParams = cacheTypeMap.get(type);
        ConsumeType consumeType = null;
        if (cacheParams != null) {
            Date time = (Date) cacheParams.get("time");
            if (DateUtil.getDistanceOfTwoMinite(time, new Date()) < 180) {//缓存时间为180分钟
                consumeType = (ConsumeType) cacheParams.get("consumeType");
            }
        }
        if (consumeType == null) {
            HashMap<String, Object> params = new HashMap<String, Object>();
            params.put("consumeTypeName", type);
            List<ConsumeType> consumeTypes = iConsumeTypeService.findList(params);
            if (consumeTypes.size() == 0) {
                return null;
            }
            consumeType = consumeTypes.get(0);
            consumeType.setListSet(iConsumeTypeService.findConsumeTypeSet(consumeType.getConsumeTypeId()));
            cacheParams = new HashMap<String, Object>();
            cacheParams.put("time", new Date());
            cacheParams.put("consumeType", consumeType);
            cacheTypeMap.put(type, cacheParams);
        }
        return consumeType;
    }
}
